package com.taoge.mqttdemo1.boot;

import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

/**
 * 接收
 */
@Slf4j
@AllArgsConstructor
@Configuration
@IntegrationComponentScan
public class MqttInboundConfiguration {

    private MqttConfiguration mqttConfig;
    private MqttPahoClientFactory factory;
    private MqttMessageReceiver mqttMessageReceiver;

    /**
     * 此处可以使用其他消息接收通道
     * Spring Integration默认的消息通道，它允许将消息发送给一个订阅者，然后阻碍发送直到消息被接收。
     *
     * @return
     */
    @Bean
    public MessageChannel mqttInBoundChannel() {
        return new DirectChannel();
    }

    /**
     * 适配器, 两个topic共用一个adapter
     * 客户端作为消费者，订阅主题，消费消息
     * QoS0，发送就不管了，最多一次；
     * QoS1，发送之后依赖MQTT规范，是否启动重传消息，所以至少一次；
     * QoS2，发送之后依赖MQTT消息机制，确保只有一次。
     *
     * QoS0 代表，Sender 发送的一条消息，Receiver 最多能收到一次，也就是说 Sender 尽力向 Receiver 发送消息，如果发送失败，也就算了；这是完全依赖TCP重传机制，如果网络不好，TCP的重传也不是100%可靠，加上MQTT是Publisher 发出去的消息是依赖代理服务器完成转发，所以消息最多一次。
     * QoS1 代表，Sender 发送的一条消息，Receiver 至少能收到一次，也就是说 Sender 向 Receiver 发送消息，如果发送之后没有收到对应的PUBACK，就会继续重试，直到发送者Sender 接收到 Receiver 发送的 PUBACK 为止，因为重传的原因，Receiver 有可能会收到重复的消息；
     * QoS2 代表，Sender 发送的一条消息，Receiver 确保能收到而且只收到一次，也就是说 Sender 尽力向 Receiver 发送消息，如果发送失败，会继续重试，直到 Receiver 收到消息为止，同时保证 Receiver 不会因为消息重传而收到重复的消息。（有点像TCP三次握手的交互过程）
     *
     * @param
     * @param
     * @return
     */
    @Bean
    public MessageProducerSupport mqttInbound() {
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter(mqttConfig.getClientId() + "-" + System.currentTimeMillis(), factory, mqttConfig.getTopic());
        adapter.setCompletionTimeout(60000);
        adapter.setConverter(new DefaultPahoMessageConverter());
//        adapter.setRecoveryInterval(10000);
        adapter.setQos(0);
        adapter.setOutputChannel(mqttInBoundChannel());
        return adapter;
    }

    /**
     * mqtt入站消息处理工具，对于指定消息入站通道接收到生产者生产的消息后处理消息的工具。
     * @ServiceActivator 该注解所标注的方法是以服务的形式出现的，只要管道中的数据发生了变化就会激活该服务。返回值可以是任何类型。如果返回值不是消息，则将以该对象作为其有效负载创建应答消息。
     * @SteamListener 该方式是以监听的方式实现，只要管道中的流数据发生变化，其就会触发该注解所标注的方法的执行。
     * @return
     */
    @Bean
    @ServiceActivator(inputChannel = "mqttInBoundChannel")
    public MessageHandler mqttMessageHandler() {
        return this.mqttMessageReceiver;
        /* return (message -> {
            String topic = message.getHeaders().get("mqtt_receivedTopic").toString();
            String payload = message.getPayload().toString();
            System.out.println("消息主题：" + topic);
            System.out.println("消息内容：" + payload);
        });*/
    }
}
